package com.sensei.search.nodes;

import java.io.File;
import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;
import org.apache.lucene.queryParser.QueryParser;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.channel.group.ChannelGroupFuture;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.ServerSocketChannelFactory;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.FileSystemXmlApplicationContext;

import proj.zoie.api.IndexReaderFactory;

import com.browseengine.bobo.api.BoboIndexReader;

public class SenseiServer {
	private static final Logger logger = Logger.getLogger(SenseiServer.class);
	private static final String DEFAULT_CONF_FILE = "sensei-node.spring";
	

	private static final String DEFAULT_ZK_URL = "localhost:2181";
	
	private ServerBootstrap _bootstrap;
	private ExecutorService _bootstrapExecutor;
	private ChannelGroup _channelGroup;
	private boolean _started;
	
	private final int _port;
	
	private final QueryParser _qparser;
	private final IndexReaderFactory<BoboIndexReader> _idxReaderFactory;
	
	SenseiServer(int port,QueryParser qparser,IndexReaderFactory<BoboIndexReader> idxReaderFactory){
		_port = port;
		_qparser = qparser;
		_idxReaderFactory = idxReaderFactory;
		_started = false;
	}
	
	public int getPort(){
		return _port;
	}
	
	void startup(){
		if (_started){
			logger.error("server already started");
			return;
		}
		_bootstrapExecutor = Executors.newCachedThreadPool();
		_channelGroup = new DefaultChannelGroup("sensei-server");
		ServerSocketChannelFactory serverChannelFactory = new NioServerSocketChannelFactory(_bootstrapExecutor,_bootstrapExecutor);
		_bootstrap= new ServerBootstrap(serverChannelFactory);
		
		_bootstrap.setOption("tcpNoDelay", true);
	    _bootstrap.setOption("reuseAddress", true);
	    _bootstrap.setOption("backlog", 1000);
	    _bootstrap.setOption("child.tcpNoDelay", true);
	    _bootstrap.setOption("child.reuseAddress", true);
	    
	    _bootstrap.setPipelineFactory(new SenseiServerPipelineFactory(_qparser,_idxReaderFactory));
	    
	    InetSocketAddress localAddr = new InetSocketAddress("0.0.0.0",_port);
	    _bootstrap.bind(localAddr);
	    
	    _started = true;
	    logger.info("listening on port: "+_port+"...");
	}
	
	void shutdown(){
		if (_started){
		  logger.info("shutting down...");
		  ChannelGroupFuture closeFuture = _channelGroup.close();
		  closeFuture.awaitUninterruptibly();
		  _bootstrap.getFactory().releaseExternalResources();
		  _started = false;
		  logger.info("senser shut down..."); 
		}
	}
	
	private static String help(){
		StringBuffer buffer = new StringBuffer();
		buffer.append("Usage: [id] [port] [partitions] [conf.dir] <zookeeper url>\n");
		buffer.append("====================================\n");
		buffer.append("id - node id (integer), required\n");
		buffer.append("port - server port (integer), required\n");
		buffer.append("partitions - comma separated list of partition numbers this node can serve, required\n");
		buffer.append("conf.dir - server configuration directory, required\n");
		buffer.append("zookeeper url - url (form: host:port) of the zookeeper instance/cluster, optional, default: ").append(DEFAULT_ZK_URL).append("\n");
		buffer.append("====================================\n");
		return buffer.toString();
	}
	
	public static void main(String[] args) throws Exception{
		if (args.length<4){
			System.out.println(help());
			System.exit(1);
		}
		
		int id = 0;
		int port = 0;
		int[] partitions = null;
		String[] partString = null;
		
		File confDir = null;
		String zookeeperURL = null;
		File confFile = null;
		
		try{
			id = Integer.parseInt(args[0]);
			port = Integer.parseInt(args[1]);
			partString = args[2].split(",");
			confDir = new File(args[3]);
			confFile = new File(confDir,DEFAULT_CONF_FILE);
			
			
			
			partitions = new int[partString.length];
			for (int i=0;i<partString.length;++i){
				partitions[i] = Integer.parseInt(partString[i]);
			}
			
			
			try{
				zookeeperURL = args[4];
			}
			catch(Exception e){
				zookeeperURL = null;
			}
			
			if (zookeeperURL == null){
				System.out.println("invalid or no cluster url specified, defaulting to default: "+DEFAULT_ZK_URL);
				zookeeperURL = DEFAULT_ZK_URL;
			}
		}
		catch(Exception e){
			System.out.println(help());
			System.exit(0);
		}
		
		ApplicationContext springCtx = new FileSystemXmlApplicationContext("file:"+confFile.getAbsolutePath());
		
		QueryParser qparser = (QueryParser)springCtx.getBean("query-parser");
		IndexReaderFactory<BoboIndexReader> idxReaderFactory = (IndexReaderFactory<BoboIndexReader>)springCtx.getBean("index-reader-factory");
		
		SenseiServer server = new SenseiServer(port,qparser,idxReaderFactory);
		final SenseiNode node = new SenseiNode(id,partitions,server,zookeeperURL);
		node.startup();
		
		
		Runtime.getRuntime().addShutdownHook(new Thread(){
			public void run(){
				try {
					node.shutdown();
				} catch (Exception e) {
					logger.error(e.getMessage(),e);
				}
			}
		});
		
		while(true){
			Thread.sleep(5000);
		}		
	}
}
